package drds.server.config.table.structure;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.config.model.Schema;
import drds.server.config.model.Table;
import drds.server.datasource.GroupDataSourceNode;
import drds.server.execute_engine.OneRawSQLQueryResultHandler;
import drds.server.execute_engine.QueryResult;
import drds.server.execute_engine.QueryResultListener;
import drds.server.execute_engine.Task;

/**
 * 表结构结果处理
 */
public class TableStructureDetector implements Runnable {
	private static final Logger LOGGER = LoggerFactory.getLogger(TableStructureDetector.class);
	private static final String[] MYSQL_SHOW_CREATE_TABLE_COLMS = new String[] { "Table", "Create Table" };
	private static final String sqlPrefix = "show create table ";

	@Override
	public void run() {

		Schema schema = Server.getInstance().getConfig().getSchema();
		for (Table table : schema.getTableMap().values()) {
			for (String dataNode : table.getDataNodeList()) {
				try {
					table.getReadWriteLock().writeLock().lock();
					ConcurrentHashMap<String, List<String>> map = new ConcurrentHashMap<>();
					table.setDataNodeTableStructureSQLMap(map);
				} finally {
					table.getReadWriteLock().writeLock().unlock();
				}
				OneRawSQLQueryResultHandler resultHandler = new OneRawSQLQueryResultHandler(MYSQL_SHOW_CREATE_TABLE_COLMS, new MySQLTableStructureListener(dataNode, table));
				resultHandler.setMark("Table Structure");
				GroupDataSourceNode dn = Server.getInstance().getConfig().getGroupDataSourceNodeMap().get(dataNode);
				Task sqlJob = new Task(sqlPrefix + table.getName(), dn.getSchemaName(), resultHandler, dn.getGroupDataSource().getWriteDataSource());
				sqlJob.run();
			}
		}
	}

	private static class MySQLTableStructureListener implements QueryResultListener<QueryResult<Map<String, String>>> {
		private String dataNode;
		private Table table;

		public MySQLTableStructureListener(String dataNode, Table table) {
			this.dataNode = dataNode;
			this.table = table;
		}

		/**
		 * @param result
		 * @// TODO: 2016/5/11 检查表元素，来确定是哪个元素不一致，未来还有其他用
		 */
		@Override
		public void onResult(QueryResult<Map<String, String>> result) {
			try {
				table.getReadWriteLock().writeLock().lock();
				if (!result.isSuccess()) {
					LOGGER.error("Can't get table " + table.getName() + "'s config from DataNode:" + dataNode + "! Maybe the table is not initialized!");
					return;
				}
				String currentSql = result.getResult().get(MYSQL_SHOW_CREATE_TABLE_COLMS[1]);
				Map<String, List<String>> dataNodeTableStructureSQLMap = table.getDataNodeTableStructureSQLMap();
				if (dataNodeTableStructureSQLMap.containsKey(currentSql)) {
					List<String> dataNodeList = dataNodeTableStructureSQLMap.get(currentSql);
					dataNodeList.add(dataNode);
				} else {
					List<String> dataNodeList = new LinkedList<>();
					dataNodeList.add(dataNode);
					dataNodeTableStructureSQLMap.put(currentSql, dataNodeList);
				}
				if (dataNodeTableStructureSQLMap.size() > 1) {
					LOGGER.error("Table [" + table.getName() + "] structure are not consistent!");
					LOGGER.error("Currently detected: ");
					for (String sql : dataNodeTableStructureSQLMap.keySet()) {
						StringBuilder stringBuilder = new StringBuilder();
						for (String dn : dataNodeTableStructureSQLMap.get(sql)) {
							stringBuilder.append("DataNode:[").append(dn).append("]");
						}
						stringBuilder.append(":").append(sql);
						LOGGER.error(stringBuilder.toString());
					}
				}
			} finally {
				table.getReadWriteLock().writeLock().unlock();
			}
		}
	}
}
